package org.example.cloudstream;

import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.cloud.stream.function.StreamBridge;
import org.springframework.context.annotation.Bean;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.HashMap;
import java.util.Map;
import java.util.function.Consumer;

@SpringBootApplication
@RestController
@RequestMapping
@Slf4j
public class CloudstreamApplication {

    public static void main(String[] args) {
        SpringApplication.run(CloudstreamApplication.class, args);
    }

    @Autowired
    private StreamBridge bridge;

//    @PostMapping("sendKafka/{topic}")
//    public ResponseEntity<Boolean> sendKafka(@PathVariable("topic") String topic, @RequestBody Map<String, Object> map) {
//        map.put("name", "张三");
//        boolean send = bridge.send("outputKafka3-out-0", map);
////        boolean send = bridge.send(topic, map);
//        return ResponseEntity.ok(send);
//    }

    @PostMapping("sendKafka")
    public ResponseEntity<Boolean> sendKafka() {
        Map<String, Object> map = new HashMap<>();
        map.put("name", "张三");
        boolean send = bridge.send("outputKafka-out-0", map);
//        boolean send = bridge.send(topic, map);
        return ResponseEntity.ok(send);
    }

    // 供应商函数，就是发送消息者，供应消息的
    // 会一直自动发送消息
//    @Bean
//    public Supplier<String> outputKafka() {
//        return () -> {
//            String message = "from outputKafka";
//            System.out.println("------------------from outputKafka--------------");
//            return message;
//        };
//    }

//    @Bean
//    public Supplier<String> outputKafka3() {
//        return () -> {
//            String message = "from outputKafka";
//            System.out.println("------------------from outputKafka--------------");
//            return message;
//        };
//    }
//
//    @Bean
//    public Supplier<String> outputKafka4() {
//        return () -> {
//            String message = "from outputKafka";
//            System.out.println("------------------from outputKafka--------------");
//            return message;
//        };
//    }


    @Bean
    public Consumer<String> inputKafka1() {
        return str -> {
            // 收到消息在这里做一些处理
            log.info("inputKafka1 message: {}", str);
        };
    }

    @Bean
    public Consumer<String> inputKafka2() {
        return str -> {
            log.info("inputKafka2 message: {}", str);
        };
    }

    @Bean
    public Consumer<String> inputKafka3() {
        return str -> {
            log.info("inputKafka3 message: {}", str);
        };
    }

    @Bean
    public Consumer<String> inputKafka4() {
        return str -> {
            log.info("inputKafka4 message: {}", str);
        };
    }
}
